package com.junbao.message.infrastructure.core.dispatcher;

import java.util.concurrent.ThreadPoolExecutor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.junbao.commom.context.ClassPathXmlApplicationContext;
import com.junbao.message.application.service.MessageService;
import com.junbao.message.domain.model.NotifyMessage;
import com.junbao.message.infrastructure.core.AbstractDisruptorCore;
import com.junbao.message.infrastructure.core.DataConsumer;
import com.junbao.message.infrastructure.core.DataProducer;
import com.junbao.message.infrastructure.core.RequestDto;
import com.junbao.message.infrastructure.core.handler.MessageSendHandler;
import com.junbao.message.infrastructure.core.producer.MessageSendProducer;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.WorkProcessor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * 消息发送diruptor
 * @author yilin
 *
 * @param <T>
 */
@SuppressWarnings("all")
public class MessageSendDisruptor extends AbstractDisruptorCore{
 
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageSendDisruptor.class);
    
    private static class DisruptorCoreInstance{
        private final static MessageSendDisruptor INSTANCE = new MessageSendDisruptor();
    }
    
    private MessageSendDisruptor(){};
    
    public static MessageSendDisruptor instance(){
        return DisruptorCoreInstance.INSTANCE;
    }
    /**
     * 消费者数量
     */
    private static final int CONSUMER_COUNT = (int) (Runtime.getRuntime().availableProcessors()/0.1);
    
    /** 最大线程数量*/
    private static final int MAX_THREAD_COUNT = (int) (Runtime.getRuntime().availableProcessors()/0.1);
    
    static{
        try {
            MessageSendDisruptor disruptorCore = MessageSendDisruptor.instance();
            //获取消费者
            DataConsumer[] messageConsumers = getmessageConsumers(disruptorCore.setCounsumerClassType(),CONSUMER_COUNT);
            //启动disruptor
            disruptorCore.start(ProducerType.SINGLE,
                    RINGBUFFER_SIZE,
                    new BlockingWaitStrategy(),
                    messageConsumers,
                    RequestDto.class);
        } catch (Exception e) {
            LOGGER.error("启动MessageSendDisruptor失败！",e);
        }
    }
 
    @Override
    public DataProducer createDataProducer(String producerName) {
        return new MessageSendProducer(producerName, this.ringBuffer);
    }
    
    @Override
    public int setCoreThreadSize() {
        return MAX_THREAD_COUNT;
    }
    
    @Override
    public int setMaxThreadSize() {
        return MAX_THREAD_COUNT;
    }
    
    @Override
    public String setWorkThreadName() {
        return "MessageSend-thread";
    }
 
    @Override
    public Class setCounsumerClassType() {
        return MessageSendHandler.class;
    }
    
    @Override
    public void rejectedExecutionHandler(Runnable runnable, ThreadPoolExecutor executor) {
        executor.execute(runnable);
    }
 
    @Override
    public void disruptorExceptionHandler(Throwable throwable, long sequence, Object event) {
        RequestDto<NotifyMessage> requestDto =(RequestDto)event;
        LOGGER.error("消息发送失败，消息编号："+requestDto.getBusinessObject().getMessageNo(),throwable);
        MessageService messageService = (MessageService) ClassPathXmlApplicationContext.getInstance().getBean("messageService");
        messageService.sendMessage(requestDto.getBusinessObject());
    }
}